package com.bw.gmall.realtime.app.dwd;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
//9.2 流量域独立访客事务事实表
public class DwdTrafficUniqueVisitorDetail {
    /**
     * 主程序入口，负责构建并执行Flink流处理任务
     *
     * @param args 命令行参数
     * @throws Exception 执行过程中可能抛出的异常
     */
    public static void main(String[] args) throws Exception {

        // TODO 1. 环境准备创建流处理执行环境并设置并行度为1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2. 状态后端设置
        // 配置检查点机制（每3秒做一次精确一次的检查点）
        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(30 * 1000L);  // 检查点超时时间30秒
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);  // 检查点最小间隔3秒
        // 配置检查点外部持久化（任务取消时保留检查点）
        env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
        );
        // 设置失败重启策略（每天最多失败3次，每次间隔1分钟）
        env.setRestartStrategy(RestartStrategies.failureRateRestart(
                3, Time.days(1), Time.minutes(1)
        ));

        // 配置状态后端和检查点存储路径
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage(
                "hdfs://hadoop102:8020/ck"  // HDFS存储路径
        );

        System.setProperty("HADOOP_USER_NAME", "bw");  // 设置Hadoop操作用户

        // TODO 3. 从 kafka topic_log 主题读取日志数据，封装为流
        String topic = "topic_log";
        String groupId = "dwd_traffic_unique_visitor_detail";
        // 创建Kafka消费者并获取数据流
        System.out.println("正在从Kafka主题" + topic + "消费数据，消费者组ID：" + groupId);
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId);
        DataStreamSource<String> pageLog = env.addSource(kafkaConsumer);
        pageLog.print("原始日志数据:");  // 打印原始日志数据
        
        // 添加计数器统计接收到的日志数量
        pageLog.map(new MapFunction<String, String>() {
            private int logCount = 0;
            @Override
            public String map(String value) throws Exception {
                logCount++;
                if (logCount % 10 == 0) {
                    System.out.println("已接收日志总数: " + logCount);
                }
                return value;
            }
        });

        // TODO 4. 转换结构
        // 将JSON字符串转换为JSONObject对象
        SingleOutputStreamOperator<JSONObject> mappedStream = pageLog.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                try {
                    JSONObject jsonObj = JSON.parseObject(s);
                    System.out.println("成功解析JSON: 包含common=" + (jsonObj.containsKey("common")) + ", page=" + (jsonObj.containsKey("page")) + ", ts=" + jsonObj.getLong("ts"));
                    return jsonObj;
                } catch (Exception e) {
                    System.out.println("JSON解析异常: " + e.getMessage() + ", 原始数据: " + s);
                    return null;
                }
            }
        });
        // 过滤掉null值
        mappedStream = mappedStream.filter(jsonObj -> jsonObj != null);

        // TODO 5. 过滤 last_page_id 不为 null 的数据
        // 保留页面跳转来源为空的首访记录
        SingleOutputStreamOperator<JSONObject> firstPageStream = mappedStream.filter(
                jsonObj -> jsonObj != null &&
                           jsonObj.getJSONObject("page") != null &&
                           jsonObj.getJSONObject("page").getString("last_page_id") == null
        );
        firstPageStream.print("首次访问页面:");  // 打印首次访问页面数据

        // TODO 6. 按照 mid 分组
        // 根据设备唯一标识mid进行分组
        KeyedStream<JSONObject, String> keyedStream = firstPageStream
                .keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        // TODO 7. 通过 Flink 状态编程过滤独立访客记录
        SingleOutputStreamOperator<JSONObject> filteredStream = keyedStream.filter(
                /**
                 * 自定义富过滤函数，实现每日独立访客去重
                 */
                new RichFilterFunction<JSONObject>() {

                    private ValueState<String> lastVisitDt;  // 状态存储上次访问日期

                    @Override
                    public void open(Configuration paramenters) throws Exception {
                        super.open(paramenters);
                        // 配置带TTL的状态（自动清理1天前的状态）
                        ValueStateDescriptor<String> valueStateDescriptor =
                                new ValueStateDescriptor<>("last_visit_dt", String.class);
                        valueStateDescriptor.enableTimeToLive(
                                StateTtlConfig
                                        .newBuilder(Time.days(1L))
                                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                        .build()
                        );
                        lastVisitDt = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    /**
                     * 过滤逻辑：判断当前访问日期是否与上次访问日期不同
                     */

                    @Override
                    public boolean filter(JSONObject jsonObj) throws Exception {
                        try {
                            String mid = jsonObj.getJSONObject("common").getString("mid");
                            long ts = jsonObj.getLong("ts");
                            String visitDt = DateFormatUtil.toDate(ts);  // 转换时间戳为日期
                            String lastDt = lastVisitDt.value();
                            
                            System.out.println("UV过滤检查 - mid: " + mid + ", 当前日期: " + visitDt + ", 上次访问日期: " + lastDt);
                            
                            // 首次访问或日期变更时保留记录
                            if (lastDt == null) {
                                System.out.println("新访客: " + mid + " 在 " + visitDt + " 首次访问");
                                lastVisitDt.update(visitDt);
                                return true;
                            } else if (!lastDt.equals(visitDt)) {
                                System.out.println("回访客: " + mid + " 从 " + lastDt + " 更新到 " + visitDt);
                                lastVisitDt.update(visitDt);
                                return true;
                            } else {
                                System.out.println("重复访问: " + mid + " 当天已访问过，已过滤");
                                return false;
                            }
                        } catch (Exception e) {
                            System.out.println("UV过滤异常: " + e.getMessage());
                            e.printStackTrace();
                            return false;
                        }
                    }
                }
        );
        filteredStream.print("独立访客数据:");  // 打印独立访客数据

        // TODO 8. 将独立访客数据写入
        // Kafka dwd_traffic_unique_visitor_detail 主题
        String targetTopic = "dwd_traffic_unique_visitor_detail";
        // 创建Kafka生产者并写入处理结果
        System.out.println("准备写入Kafka主题: " + targetTopic);
        FlinkKafkaProducer<String> kafkaProducer = MyKafkaUtil.getFlinkKafkaProducer(targetTopic);
        filteredStream.map(new MapFunction<JSONObject, String>() {
            private int writeCount = 0;
            @Override
            public String map(JSONObject jsonObj) throws Exception {
                writeCount++;
                String mid = jsonObj.getJSONObject("common").getString("mid");
                String result = jsonObj.toJSONString();
                System.out.println("即将写入Kafka的数据 " + writeCount + ": mid=" + mid);
                return result;
            }
        }).addSink(kafkaProducer);
        System.out.println("Kafka写入sink已添加完成");

        // TODO 9. 启动任务
        System.out.println("DwdTrafficUniqueVisitorDetail任务启动中...");
        env.execute("DwdTrafficUniqueVisitorDetail");
    }
}

